package com.atguigu.distributed.lock.zookeeper.lock;

import com.atguigu.distributed.lock.redis.lock.DistributedRedisLock;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.*;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;

/**
 * @author linjy on 2022/10/19
 */
public class ZkDistributedLock implements Lock {

    private static final String ROOT_PATH = "/distributed_locks_test";

    private String lockName;

    private ZooKeeper zooKeeper;

    private String currentNodePath;

    private static final ThreadLocal<Integer> THREAD_LOCAL = new ThreadLocal<>();

    public ZkDistributedLock(ZooKeeper zooKeeper, String lockName) {
        this.zooKeeper = zooKeeper;
        this.lockName = lockName;
        try {
            if (zooKeeper.exists(ROOT_PATH, false) == null) {
                zooKeeper.create(ROOT_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void lock() {
        this.tryLock();
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    /**
     * 优化阻塞锁
     * 注意，这里的THREAD_LOCAL是针对同一把zookeeper锁（getZkDistributedLock("lock")），不同的锁会失效
     * @return
     */
    @Override
    public boolean tryLock() {
        try {
            //判断threadLocal中是否已经有锁，有锁直接重入（+1）
            Integer flag = THREAD_LOCAL.get();
            if (flag != null && flag > 0) {
                THREAD_LOCAL.set(flag + 1);
                return true;
            }

            // 创建znode节点过程:为了防止zk客户端获取到锁之后，服务器宕机带来的死锁问题，这里创建的是临时节点
            currentNodePath = zooKeeper.create(ROOT_PATH + "/" + lockName + "-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // 获取前置节点，如果前置节点为空，则获取锁成功，否则监听前置节点(公平锁)
            String preNode = this.getPreNode();
            if (preNode != null) {
                //利用闭锁思想，实现阻塞功能
                CountDownLatch countDownLatch = new CountDownLatch(1);
                //再次判断zk中的前置节点是否存在（因为获取前置节点这个操作，不具备原子性）
                if (this.zooKeeper.exists(ROOT_PATH + "/" + preNode, new Watcher() {
                    @Override
                    public void process(WatchedEvent watchedEvent) {
                        countDownLatch.countDown();
                    }
                }) == null) {
                    //别的线程或者别的进程获取到锁，释放后，此线程获取到锁
                    THREAD_LOCAL.set(1);
                    return true;
                }
                countDownLatch.await();
            }
            //只有当前线程首次加锁,因为最前面的判断了【if (flag != null && flag > 0)】已经重入的锁
            THREAD_LOCAL.set(1);
            return true;
        } catch (Exception e) {
            // 重试
            try {
                Thread.sleep(200);
                tryLock();
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
        }
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        try {
            THREAD_LOCAL.set(THREAD_LOCAL.get() - 1);
            if (THREAD_LOCAL.get() == 0) {
                //删除znode节点过程
                this.zooKeeper.delete(currentNodePath, -1);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    /**
     * 获取指定节点的前节点
     */
    private String getPreNode() {

        try {
            // 获取根路径下的所有序列化子节点
            List<String> childrenList = this.zooKeeper.getChildren(ROOT_PATH, false);
            // 判空
            if (CollectionUtils.isEmpty(childrenList)) {
                throw new IllegalMonitorStateException("非法操作");
            }

            //获取和当前节点同一资源的锁
            List<String> nodesList = childrenList.stream().filter(node -> StringUtils.startsWith(node, lockName + "-")).collect(Collectors.toList());
            // 判空
            if (CollectionUtils.isEmpty(nodesList)) {
                throw new IllegalMonitorStateException("非法操作");
            }

            //排好队
            Collections.sort(nodesList);

            //获取当前节点的下标
            String currentNode = StringUtils.substringAfterLast(currentNodePath, "/");
            int index = Collections.binarySearch(nodesList, currentNode);
            if (index < 0) {
                throw new IllegalMonitorStateException("非法操作");
            } else if (index > 0) {
                return nodesList.get(index - 1);//返回前置节点
            }
            //如果当前节点就是第一个节点，则返回null
            return null;

        } catch (Exception e) {
            e.printStackTrace();
            throw new IllegalMonitorStateException("非法操作");
        }

    }
}
